#include "stdafx.h"
#include "ORB_Notify_Supplier.h"
#include "tao/ORB_Core.h"

ORB_Notify_Supplier::ORB_Notify_Supplier(void) :
proxy_consumer_id_ (0),
proxy_consumer_thread_count_ (0)
{
}

ORB_Notify_Supplier::~ORB_Notify_Supplier()
{
    //// Disconnect from the EC
    //this->disconnect ();

    //// Deactivate this object.
    //this->deactivate ();
}

void
ORB_Notify_Supplier::init_supplier(CosNotifyChannelAdmin::SupplierAdmin_var& admin,
                          int proxy_consumer_thread_count)
{
    // First initialize the class members.
    this->admin_ = admin;
    this->proxy_consumer_thread_count_ = proxy_consumer_thread_count;
    
    this->connect();
}

void
ORB_Notify_Supplier::connect (void)
{
    // Activate the supplier object.
    CosNotifyChannelAdmin::ProxyConsumer_var proxyconsumer;
    if (this->proxy_consumer_thread_count_ != 0){
        // Narrow to the extended interface.
        NotifyExt::SupplierAdmin_var admin_ext = NotifyExt::SupplierAdmin::_narrow (this->admin_.in());
        NotifyExt::ThreadPoolParams tp_params = { NotifyExt::CLIENT_PROPAGATED, 0,
            0, this->proxy_consumer_thread_count_, 0, 0, 0, 0, 0 };

        CosNotification::QoSProperties qos (1);
        qos.length (1);
        qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
        qos[0].value <<= tp_params;

        // Obtain the proxy. The QoS is applied to the POA in which the Proxy is hosted.
        proxyconsumer = admin_ext->obtain_notification_push_consumer_with_qos (
            CosNotifyChannelAdmin::STRUCTURED_EVENT, proxy_consumer_id_, qos);
    }
    else{
        // Obtain the proxy.
        proxyconsumer = this->admin_->obtain_notification_push_consumer (
            CosNotifyChannelAdmin::STRUCTURED_EVENT, proxy_consumer_id_);
    }

    ACE_ASSERT (!CORBA::is_nil (proxyconsumer.in()));

    // narrow
    this->proxy_consumer_ = CosNotifyChannelAdmin::StructuredProxyPushConsumer::_narrow (proxyconsumer.in());
    ACE_ASSERT (!CORBA::is_nil (proxy_consumer_.in()));

    // connect to the proxyconsumer.
    this->proxy_consumer_->connect_structured_push_supplier(this->_this());

    ACE_DEBUG ((LM_DEBUG, "Created ORB_Notify_Supplier %d with %d threads.\n",
        proxy_consumer_id_,this->proxy_consumer_thread_count_));
}

void
ORB_Notify_Supplier::disconnect (void)
{
    ACE_ASSERT (!CORBA::is_nil (this->proxy_consumer_.in()));
    this->proxy_consumer_->disconnect_structured_push_consumer();
}

void
ORB_Notify_Supplier::deactivate (void)
{
    PortableServer::POA_var poa (this->_default_POA());
    PortableServer::ObjectId_var id (poa->servant_to_id (this));
    poa->deactivate_object (id.in());
}

void
ORB_Notify_Supplier::send_event (const CosNotification::StructuredEvent& event)
{
    if(!CORBA::is_nil (this->proxy_consumer_.in()))
        proxy_consumer_->push_structured_event (event);
}

void
ORB_Notify_Supplier::subscription_change (const CosNotification::EventTypeSeq & added,
                                          const CosNotification::EventTypeSeq & removed )
{
}

void
ORB_Notify_Supplier::disconnect_structured_push_supplier (void)
{
    this->deactivate();
}
